package com.clj.rocketmq_example.mq.listener;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.common.message.MessageExt;

import com.clj.rocketmq_example.annotation.RocketMqListener;
import com.clj.rocketmq_example.enums.MqAction;
import com.clj.rocketmq_example.mq.MessageListener;
import com.clj.rocketmq_example.mq.command.MqConstant;
import com.clj.rocketmq_example.util.MqMsgConvertUtil;

/**
 * 普通消息监听
 **/
@RocketMqListener(topic = MqConstant.ORDER_TABLE_TOP,consumerGroup= MqConstant.ORDER_GROUP)
public class MqListenerTwo implements MessageListener{

	private static Logger LOGGER = LogManager.getLogger();

	@Override
	public MqAction consume(MessageExt message, ConsumeConcurrentlyContext context) {
		String msg=null;
		try {
			System.out.println("消费者===========");
			msg=MqMsgConvertUtil.bytes2String(message.getBody(), "utf8");
			LOGGER.info("MsgId:{},MQ消费,Topic:{},Tag:{}，Body:{}",message.getMsgId(),message.getTopic(),message.getTags(),message.getBody());
		} catch (Exception e) {
			 LOGGER.error("MsgId:{},应用MQ消费失败,Topic:{},Tag:{}，Body:{},异常信息:{}", message.getMsgId(), message.getTopic(), message.getTags(), msg, e);
	            throw e;
	        }
	        return MqAction.CommitMessage;
	}

	
	
	
}
